package com.itcast.flink.connectors.jdbc;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @program: flink-app
 * @description: 从mysql中读取数据
 * @author: zhanghz001
 * @create: 2021-07-23 09:51
 **/
public class ZhzMySQLSourceFunction extends RichSourceFunction<AccessLog> {
    /**
     * 数据库连接
     */
    private Connection connection;
    /**
     * 预编译sql
     */
    
    private PreparedStatement preparedStatement;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        String url = "jdbc:mysql://192.168.23.140:3306/flink?useSSL=false";
        String executionSql = "select id, ip, time, type, api from t_access_log;";
        connection = DriverManager.getConnection(url, "root", "654321");
        preparedStatement = connection.prepareStatement(executionSql);
    }
    
    @Override
    public void run(SourceContext<AccessLog> sourceContext) throws Exception {
        //查询数据库的数据,组装返回
        ResultSet rs = preparedStatement.executeQuery();
        while (rs.next()) {
            AccessLog accessLog = new AccessLog();
            accessLog.setIp(rs.getString("ip"));
            accessLog.setTime(rs.getString("time"));
            accessLog.setType(rs.getString("type"));
            accessLog.setApi(rs.getString("api"));
            sourceContext.collect(accessLog);
        }
    }
    
    @Override
    public void close() throws Exception {
        //    释放数据库连接
        if (null != connection) {
            try {
                connection.close();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                connection = null;
            }
        }
    }
    
    @Override
    public void cancel() {
        System.out.println("cancel method ...");
    }
}
